package com.lm.flink.log;

import com.lm.flink.entry.LogEvent;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.util.Collector;

public class LogParser implements FlatMapFunction<String, LogEvent> {
    @Override
    public void flatMap(String value, Collector<LogEvent> out) throws Exception {
        try {
            // 假设日志格式: {"level":"INFO","service":"user-service","message":"User login","timestamp":1645678900000}
            // 简单解析JSON（实际生产环境建议使用Jackson）
            String[] parts = value.replace("{", "").replace("}", "").replace("\"", "").split(",");

            String level = null;
            String service = null;
            String message = null;
            long timestamp = 0;

            for (String part : parts) {
                String[] keyValue = part.split(":");
                if (keyValue.length == 2) {
                    String key = keyValue[0].trim();
                    String val = keyValue[1].trim();

                    switch (key) {
                        case "level":
                            level = val;
                            break;
                        case "service":
                            service = val;
                            break;
                        case "message":
                            message = val;
                            break;
                        case "timestamp":
                            timestamp = Long.parseLong(val);
                            break;
                    }
                }
            }

            if (level != null && service != null) {
                out.collect(new LogEvent(level, service, message, timestamp));
            }
        } catch (Exception e) {
            System.err.println("Failed to parse log: " + value + ", error: " + e.getMessage());
        }
    }
}
